<?php
/**
 * Created by PhpStorm.
 * User: guokunsheng
 * Date: 2019/04/31
 * Time: 11:50
 */

namespace app\common\lib;

use \RdKafka\Conf;
use \RdKafka\KafkaConsumer;
use \RdKafka\Producer;
use \RdKafka\TopicConf;
use \think\Config;

/**
 * kafka扩展类
 * Class Kafka
 * @package app\common\lib
 */
class Kafka
{
    private static $handler = null;
    private static $_instance = null;
    private static $option = [
        'sasl_plain_username' => 'devKafka',
        'sasl_plain_password' => 'devKafka123456',
        'bootstrap_servers' => "47.94.157.172:9093,39.96.27.157:9093,39.107.243.211:9093",
        'topic_name' => 'dev',
        'consumer_id' => 'dev',

    ];
    private static $partition = 0;

    /**
     * 初始化配置
     * Kafka constructor.
     * @param array $options
     */
    private function __construct($options = [])
    {
        self::$handler = new Conf();
        self::$handler->set('sasl.mechanisms', 'PLAIN');
        self::$handler->set('api.version.request', 'true');
        self::$handler->set('sasl.username', $options['sasl_plain_username']);
        self::$handler->set('sasl.password', $options['sasl_plain_password']);
        self::$handler->set('security.protocol', 'SASL_SSL');
        self::$handler->set('ssl.ca.location', ROOT_PATH . '/ca-cert.pem');
        self::$handler->set('group.id', $options['consumer_id']);
        self::$handler->set('message.send.max.retries', 5);
        self::$handler->set('metadata.broker.list', $options['bootstrap_servers']);
        self::$handler->set('bootstrap.servers', $options['bootstrap_servers']);
    }

    /**
     * 实例化对象
     * @return Kafka|null 对象
     */
    public static function getInstance()
    {
        if (!(self::$_instance instanceof self))
        {
            /**
             * 从配置文件中获取kafka配置
             */
            $options = array_merge(self::$option, Config::get('rdkafka'));

            self::$_instance = new self($options);
        }
        return self::$_instance;
    }

    /**
     * 生产
     * @param array $data 需要写入队列的json字符串数据
     * @param null $key 标示key
     * @return bool
     */
    public static function producer($data = [], $key = null)
    {
        $rk = new Producer(self::$handler);

        $topic = $rk->newTopic(self::$option['topic_name']);

        $topic->produce(self::$partition, 0, json_encode($data, true), $key);

        return true;
    }

    /**
     * 消费
     * @throws \RdKafka\Exception
     */
    public static function consumer()
    {
        $topicConf = new TopicConf();

        $topicConf->set('auto.offset.reset', 'smallest');

        self::$handler->setDefaultTopicConf($topicConf);

        self::$handler->setRebalanceCb(function (KafkaConsumer $kafka, $err, array $partitions = null)
        {
            switch ($err)
            {
                case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                    echo "Assign: ";
                    var_dump($partitions);
                    $kafka->assign($partitions);
                    break;
                case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
                    echo "Revoke: ";
                    var_dump($partitions);
                    $kafka->assign(NULL);
                    break;
                default:
                    throw new \Exception($err);
            }
        });


        $consumer = new KafkaConsumer(self::$handler);
        // 订阅
        $consumer->subscribe([self::$option['topic_name']]);

        while (true)
        {
            $message = $consumer->consume(480 * 1000);
            switch ($message->err)
            {
                case RD_KAFKA_RESP_ERR_NO_ERROR:

                    print_r('队列消费：' . var_export($message, true) . PHP_EOL);
                    print_r('队列数据：' . var_export(json_decode($message->payload, true), true) . PHP_EOL);


                    \think\Log::write(var_export($message, true), 'info', true);


                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    print_r("No more messages; will wait for more" . PHP_EOL);
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    print_r("Timed out" . PHP_EOL);
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
                    break;
            }
        }
    }
}
